package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.article.config.stream.KafkaStreamListener;
import com.heima.common.constants.message.ArticleVisitStreamMess;
import com.heima.common.constants.message.HotArticleConstants;
import com.heima.common.constants.message.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KeyValueMapper;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;
import org.apache.kafka.streams.kstream.Windowed;
import org.springframework.stereotype.Component;

import java.time.Duration;
import java.util.Arrays;

/**
 * @Description:
 * @Version: V1.0
 */
@Component
@Slf4j
public class HotArticleStreamHandler implements KafkaStreamListener<KStream<String, String>> {

    /**
     * 接收所有的行为数据的topic
     * @return
     */
    @Override
    public String listenerTopic() {
        return HotArticleConstants.HOTARTICLE_SCORE_INPUT_TOPIC;
    }

    /**
     * 处理后的结果发送topic
     * @return
     */
    @Override
    public String sendTopic() {
        return HotArticleConstants.HOTARTICLE_INCR_HANDLE_OUPUT_TOPIC;
    }

    // UpdateArticleMess 接收消息数据

    /**
     * 计算当前文章 点赞 阅读 评论 收藏的次数
     * @param stream
     * @return ArticleVisitStreamMess 结果：文章id=222 点赞1 阅读1 收藏1 评论1
     *
     * 1:LIKES     2:LIKES
     * 1:VIEWS
     * 1:COLLECTION
     * 1:COMMENT
     */
    @Override
    public KStream<String, String> getService(KStream<String, String> stream) {

      return stream.flatMapValues(new ValueMapper<String, Iterable<String>>() {
            @Override
            public Iterable<String> apply(String value) {
                log.info("receive message value:{}",value);
                UpdateArticleMess articleMess = JSON.parseObject(value, UpdateArticleMess.class);
                return Arrays.asList(articleMess.getArticleId()+":"+articleMess.getType().name());
            }
        })
                .groupBy(new KeyValueMapper<String, String, String>() {
                    @Override
                    public String apply(String key, String value) {
                        return value;
                    }
                })
                .windowedBy(TimeWindows.of(Duration.ofSeconds(5)))
                .count()
                .toStream()
                .map(new KeyValueMapper<Windowed<String>, Long, KeyValue<String, String>>() {
                    @Override
                    public KeyValue<String, String> apply(Windowed<String> key, Long value) {
                        // key: 1:LIKES
                        // value: 次数
                        //发送对象： ArticleVisitStreamMess json
                        //key-->mess.getArticleId()+":"+mess.getType()
                        //value  long类型  count之后的数值
                        log.info("处理后结果key:{}-----value:{}",key.key().toString(),value.toString());
                        return new KeyValue<String, String>(key.key().toString(),
                                formatObject(key.key().toString(), value));
                    }

                });

    }


    /**
     * 封装发送的数据
     * @param key   1:LIKES
     * @param value 次数
     * @return
     */
    private String formatObject(String key, Long value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();

        String[] split = key.split(":");
        if (split.length > 0) {
            mess.setArticleId(Long.valueOf(split[0]));

            // 点赞数
            if (UpdateArticleMess.UpdateArticleType.LIKES.name().equals(split[1])) {
                mess.setLike( value );
            }
            // 阅读数
            if (UpdateArticleMess.UpdateArticleType.VIEWS.name().equals(split[1])) {
                mess.setView( value );
            }
            // 评论数
            if (UpdateArticleMess.UpdateArticleType.COMMENT.name().equals(split[1])){
                mess.setComment( value );
            }
            // 收藏数
            if (UpdateArticleMess.UpdateArticleType.COLLECTION.name().equals(split[1])) {
                mess.setCollect(value );
            }

            log.info("formatObj aggs score ArticleVisitStreamMess:{}", mess);

            return JSON.toJSONString(mess);
        }

        return null;
    }
}
